RSS非対応なページの更新状況をLambda+DynamoDBで毎日確認してみた

私はこの仕組みを作ってからAWSのリリースノートが確認し放題になりました

#AWS Lambda

#AWS

#Amazon DynamoDB

#CloudWatch Events

#Amazon SNS

荒平祐次(arap)

2024.02.29

こんにちは、AWS事業本部の荒平(@0Air)です。

皆さんはインターネット上で「このページの更新を追っかけたいのにRSS対応してない・・・！」となったことはありませんか？
今回はそうなってしまった自分のためにページの更新状況を確認する仕組みを作ってみました。
（RSS非対応を追う仕組みを持つプラグインやサービスはいくつか存在していますが、しっくり来なかったため）

ちなみに、RSSに対応しているページの場合は以下のような仕組みで実現が可能です。

前置き

この仕組みは、実際に取得リクエストが送信されるため高頻度で実施しないようにしてください（相手のサーバに負荷を与える行為と判定される可能性があります）
ページを取得してハッシュ化し、差分チェックを行うという仕組み上、ページのレスポンスが毎回異なるサイトなどには利用できません（ハッシュ値が毎回異なるため）

構成図

本エントリの構成図です。
CloudWatch EventsからLambda関数を定期実行し、ウェブサイトを確認、ハッシュ化したものをDynamoDBに格納します。差分があれば、通知するといった仕組みです。

作ってみる

1. ハッシュ格納用のDynamoDB

DynamoDBのテーブルを作成します。
テーブル名は何でもよいですが、ここではWebsiteHashesとします。パーティションキーはURL(String)としました。

テーブルがアクティブになっていれば準備完了です。

2. 通知先のAmazon SNS

通知先のSNSトピックを作成し、Slackのチャンネル宛にメールが飛ぶように設定を行いました。
SNSトピックのARNはLambda関数に組み込むため、メモなどに控えておきます。

3. サイトチェック用Lambda関数

Python 3.xのランタイムを持つLambda関数を用意し、以下のコードを貼り付けます。(筆者環境はpython 3.11)
ここでは省略しますが、 requests のレイヤーを追加する必要があります。(参考)

import json
import hashlib
import requests
import boto3
from botocore.exceptions import ClientError

dynamodb = boto3.resource('dynamodb')
sns = boto3.client('sns')
TABLE_NAME = 'WebsiteHashes'
SNS_TOPIC_ARN = 'arn:aws:sns:ap-northeast-1:0000000000000:arap-slack'

URLS = [
    "https://docs.aws.amazon.com/ja_jp/mgn/latest/ug/mgn-release-notes.html",
    "https://docs.aws.amazon.com/drs/latest/userguide/drs-release-notes.html"
]

def lambda_handler(event, context):
    changes_detected = []
    for URL in URLS:
        # URLからコンテンツを取得
        response = requests.get(URL)
        content = response.text
        # コンテンツをハッシュ化
        current_hash = hashlib.sha256(content.encode()).hexdigest()
        # DynamoDBから前回のハッシュを取得
        table = dynamodb.Table(TABLE_NAME)
        try:
            response = table.get_item(Key={'URL': URL})
            previous_hash = response['Item']['hash'] if 'Item' in response else None
        except ClientError as e:
            print(e.response['Error']['Message'])
            previous_hash = None
        # ハッシュを比較
        print("Hashed:", current_hash, previous_hash)
        if current_hash != previous_hash:
            # ハッシュが異なる場合は、SNSに通知
            message = f'Website content at {URL} has changed.'
            sns.publish(TopicArn=SNS_TOPIC_ARN, Message=message)
            # 新しいハッシュをDynamoDBに保存
            table.put_item(Item={'URL': URL, 'hash': current_hash})
            changes_detected.append(URL)
    if changes_detected:
        return {
            'statusCode': 200,
            'body': json.dumps(f'Changes detected in: {", ".join(changes_detected)}; notifications sent.')
        }
    else:
        # ハッシュが同じ場合は何もしない
        return {
            'statusCode': 200,
            'body': json.dumps('No change in website content for all URLs.')
        }

9行目は作成したDynamoDBのテーブル名を、10行目は通知先のSNSトピック名をそれぞれ置き換えます。

12行目〜15行目については、対象のURLを記載します。
（余談ですが、今回私は、AWS MGNのリリースノートがRSS非対応ページのため、これを確認する目的でこの通知を構築しています。）
ついでではありますが、最近注目していたAWS DRSのリリースノートも対象に入れておきます。

続いて、Lamdba関数の権限を編集します。今回は、以下のインラインポリシーをLambda関数に追加しました。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "VisualEditor0",
            "Effect": "Allow",
            "Action": [
                "sns:Publish",
                "dynamodb:PutItem",
                "dynamodb:DescribeTable",
                "dynamodb:ListTables",
                "dynamodb:DeleteItem",
                "dynamodb:GetItem",
                "dynamodb:UpdateItem"
            ],
            "Resource": "*"
        }
    ]
}

手動でLambda関数を起動してみます。1回目はDynamoDBにハッシュが格納されていないため、通知が飛び、2回目以降は（更新がなければ）何も起こらないはずです。